package com.etlmaster.executor.steps.kafka;

import java.util.ArrayList;
import java.util.List;

import org.apache.kafka.clients.producer.Producer;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.realtime.kafka.KafkaHelper;
import com.etlmaster.executor.utils.StepState;
/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("send2Kafka")
public class SendStringListToKafkaStep extends Step{

	private SendStringListToKafkaStepMeta meta;
	public SendStringListToKafkaStep(StepMeta meta) {
		super(meta);
		this.meta=(SendStringListToKafkaStepMeta) meta;
	}

	@Override
	protected int handle()  {
		
		Producer producer;
		String topic=meta.getTopic();
		try {
			int partition=Integer.parseInt(meta.getPartition());
			List<String> messageList=null;
			if(meta.getMessageType().equalsIgnoreCase("list"))
				messageList=(List) meta.getVariable(meta.getListName());
			else
			{
				messageList=new ArrayList();
				messageList.add(meta.getMessage());
			}
			producer=KafkaHelper.getProducer();
			KafkaHelper.sendAndCheckResult(topic,partition,0,producer,messageList);
		} catch (Exception e) {
			meta.log("ERROR","发送消息异常：{}",e.getMessage());
			return StepState.RUNNING_FAILED;
		}
		meta.log("INFO","消息发送完成");
		return StepState.RUNNING_SUCCESSED;
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
